package generic.rabbitmq;

import generic.Subscriber;

import java.io.IOException;

import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConsumerCancelledException;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.client.ShutdownSignalException;

public class RabbitMqSubscriber {

	private Subscriber listener;

	private String url = "localhost";
	private int port = 5672;

	//private final static String QUEUE_NAME = "hello";
	private static final String EXCHANGE_NAME = "logs";

	private StringBuilder builder;

	public RabbitMqSubscriber(Subscriber listener) {
		super();
		this.listener = listener;

		builder = new StringBuilder();

		initialize(url);

		System.out.println("RabbitMqListener waiting for messages...");
	}

	private void initialize(String brokerUrl) {
		try {
			ConnectionFactory factory = new ConnectionFactory();
			factory.setHost(url);
			factory.setPort(port);
			Connection connection = factory.newConnection();
			Channel channel = connection.createChannel();
			
			//channel.queueDeclare(QUEUE_NAME, false, false, false, null);
			channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
			String queueName = channel.queueDeclare().getQueue();
	        channel.queueBind(queueName, EXCHANGE_NAME, "");

			QueueingConsumer consumer = new QueueingConsumer(channel);
			channel.basicConsume(queueName, true, consumer);

			while (true) {
				QueueingConsumer.Delivery delivery = consumer.nextDelivery();
				String message = new String(delivery.getBody());
				listener.receive(message);
			}
		} catch (IOException e) {
			e.printStackTrace();
		} catch (ShutdownSignalException e) {
			e.printStackTrace();
		} catch (ConsumerCancelledException e) {
			e.printStackTrace();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}
}
